package mq

import (
	log "cyiris/core/zaplog"
	"time"

	"github.com/streadway/amqp"
)

var connMap map[string]*Connector

/**
 * MQ 链接
 */
type Connector struct {
	name            string
	addr            string
	connection      *amqp.Connection
	done            chan bool
	notifyConnClose chan *amqp.Error
	isReady         bool
}

const (
	// When reconnecting to the server after connection failure
	reconnectDelay = 5 * time.Second
)

func NewConnector(name, addr string) (conn *Connector) {
	var (
		ok bool
	)

	if conn, ok = connMap[name]; !ok {
		conn = &Connector{
			name: name,
			addr: addr,
			done: make(chan bool),
		}

		go conn.handleReconnect(addr)

		if nil == connMap {
			connMap = make(map[string]*Connector)
		}

		connMap[name] = conn
	}

	return conn
}

/**
 * 尝试重连
 */
func (conn *Connector) handleReconnect(addr string) {
	for {
		conn.isReady = false
		log.Debugf("Attempting to connect %v", addr)

		err := conn.connect(addr)

		if err != nil {
			log.Warnf("Failed to connect %v. Retrying...", addr)

			select {
			case <-conn.done:
				return
			case <-time.After(reconnectDelay):
			}
			continue
		}

		select {
		case <-conn.done:
			return
		case <-conn.notifyConnClose:
			log.Warnf("Connection closed. Reconnecting... %v", addr)
			continue
		}
	}
}

/**
 * 链接 MQ 服务器
 */
func (conn *Connector) connect(addr string) error {
	connection, err := amqp.Dial(addr)

	if err != nil {
		return err
	}

	conn.changeConnection(connection)
	log.Infof("Connected: %v", addr)
	return nil
}

func (conn *Connector) changeConnection(connection *amqp.Connection) {
	conn.connection = connection
	conn.notifyConnClose = make(chan *amqp.Error)
	conn.connection.NotifyClose(conn.notifyConnClose)
	conn.isReady = true
}
